package cn.edu.laud.sany.datatest;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.sql.SQLException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import cn.edu.laud.sany.utils.Config;
import cn.edu.laud.sany.utils.MyLog;

public class LaUDReadTest
{
	protected static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
	public static String[] IP = new String[] {};
	public static int laudPort = Config.InitLaudPort;
	public static String[] host = new String[] {};
	public static String keyspace = "";
	public static int thread = 1;
	public static int testType = 1;
	public static int querySubCount = 100;
	public static Integer[] cFS;
	public static String[] equips;
	public static Date[] startQueryTimes;
	public static Date[] endQueryTimes;
	public static String cfsFile = "";
	public static String equipFile = "";
	public static String querydateFile = "";
	public static String logFileFolder = "log";
	public static String logFileName = "readtestlog.txt";
	public static String resultLogFileName = "";
	public static long testRepeatNumber = 1;
	public static long oriDataCount = 0;
	public static Date startGetQueryDate;
	public static Date endGetQueryDate;
	public static boolean columnReversed = true;
	
	public static MyLog logger;
	public static String cqlSelectFormatString = "select %s from cf_gk_%d where equip = '%s'; ";
	public static String cqlSelectWhereFormatString = "select %s from cf_gk_%d where %s; ";
	
	public static String getLogFilePath(String fileName)
	{
		return logFileFolder + "/" + fileName;
	}
	
	public static void main(String[] args) throws ClassNotFoundException,
			SQLException, IOException
	{
		File logFolderFile = new File(logFileFolder);
		if (logFolderFile.exists() == false)
		{
			logFolderFile.mkdir();
		}
		logger = MyLog.getMyLog(getLogFilePath(logFileName), true);
		
		try
		{
			startGetQueryDate = sdf.parse("2011-10-01");
			endGetQueryDate = sdf.parse("2013-04-01");
		}
		catch (ParseException e)
		{
		}
		
		if (args.length % 2 == 1 || args.length < 1)
		{
			System.err.println(Config.ReadTestCommandHelpString);
			return;
		}
		try
		{
			parseParams(args);
		}
		catch (Exception e)
		{
			System.err.println(Config.ReadTestCommandHelpString);
			return;
		}
		
		if (testType < 1 || testType > 3)
		{
			System.err.println("testtype can only be 1, 2 or 3.");
			return;
		}
		if (getCFNames() == false)
		{
			System.err.println("can not read CFs correctly.");
			return;
		}
		if (getEquipIDs() == false)
		{
			System.err.println("can not read equip Ids correctly.");
			return;
		}
		if (testType == 1 && getQueryDateTimes() == false)
		{
			System.err.println("can not read query dates correctly.");
			return;
		}
		
		if (testType == 1)
		{
			resultLogFileName = String.format("largetablequery_%d_%d.csv",
					oriDataCount, thread);
		}
		else if (testType == 2)
		{
			resultLogFileName = String.format("hottablequery_%d_%d_2.csv",
					oriDataCount, thread);
		}
		else if (testType == 3)
		{
			resultLogFileName = String.format("hottablequery_%d_%d_3.csv",
					oriDataCount, thread);
		}
		else
		{
			System.err.println("test type must be 1, 2 or 3.");
			return;
		}
		
		long starttime = System.nanoTime();
		
		try
		{
			Class.forName("org.apache.cassandra.cql.jdbc.CassandraDriver");
		}
		catch (ClassNotFoundException e)
		{
			e.printStackTrace();
			return;
		}
		catch (Exception e)
		{
			e.printStackTrace();
			return;
		}
		
		if (testType == 1)
		{
			logger.info("create threads to do large table read test...");
		}
		else
		{
			logger.info("create threads to do hot table read test...");
		}
		
		BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
		
		ThreadPoolExecutor executor = new ThreadPoolExecutor(thread, thread,
				100, TimeUnit.DAYS, queue);
		
		logger.info("begin to run threads to import....");
		
		for (int i = 0; i < thread; i++)
		{
			String jdbcString = host[i % host.length];
			executor.execute(new LaUDTest_JDBC(i, jdbcString, keyspace,
					getLogFilePath(logFileName), testType));
		}
		executor.shutdown();
		try
		{
			executor.awaitTermination(100, TimeUnit.DAYS);
		}
		catch (InterruptedException e1)
		{
			logger.error("main thread interrupted while waiting: "
					+ e1.getMessage());
		}
		
		long endTime = System.nanoTime();
		logger.info("test program's run time(total):"
				+ cn.edu.laud.sany.utils.TimeUtils
						.nanoSecondToString((endTime - starttime)));
	}
	
	protected static void parseParams(String[] args)
	{
		for (int i = 0; i < args.length; i += 2)
		{
			if (args[i].equals("--thread"))
			{
				thread = Integer.valueOf(args[i + 1]);
			}
			else if (args[i].equals("--laudport"))
			{
				laudPort = Integer.valueOf(args[i + 1]);
			}
			else if (args[i].equals("--laudip"))
			{
				IP = args[i + 1].split(",");
				host = new String[IP.length];
				for (int k = 0; k < IP.length; k++)
				{
					host[k] = String.format(Config.JDBCFormat, IP[k], laudPort);
					logger.debug(host[k]);
				}
			}
			else if (args[i].equals("--keyspace"))
			{
				keyspace = args[i + 1];
			}
			else if (args[i].equals("--equipfile"))
			{
				equipFile = args[i + 1];
			}
			else if (args[i].equals("--cfsfile"))
			{
				cfsFile = args[i + 1];
			}
			else if (args[i].equals("--testtype"))
			{
				testType = Integer.valueOf(args[i + 1]);
			}
			else if (args[i].equals("--querydate"))
			{
				querydateFile = args[i + 1];
			}
			else if (args[i].equals("--queryhot"))
			{
				querySubCount = Integer.valueOf(args[i + 1]);
			}
			else if (args[i].equals("--repeat"))
			{
				testRepeatNumber = Long.valueOf(args[i + 1]);
			}
			else if (args[i].equals("--oridata"))
			{
				oriDataCount = Long.valueOf(args[i + 1]);
			}
			
		}
		logger.info("thread = " + thread);
		logger.info("laud port = " + laudPort);
		String ipsString = "";
		for (String ip : IP)
		{
			ipsString += ip + "; ";
		}
		logger.info("laud ip = " + ipsString);
		logger.info("keyspace = " + keyspace);
		logger.info("equips file = " + equipFile);
		logger.info("cfs file = " + cfsFile);
		logger.info("test type = " + testType);
		logger.info("query date time file = " + querydateFile);
		logger.info("sub query for hot table = " + querySubCount);
		logger.info("test repeat number = " + testRepeatNumber);
		logger.info("ori Data Count = " + oriDataCount);
	}
	
	protected static boolean getCFNames()
	{
		boolean flag = true;
		File file = new File(cfsFile);
		BufferedReader bReader = null;
		String paraIdString = null;
		ArrayList<Integer> cfList = new ArrayList<Integer>();
		try
		{
			bReader = new BufferedReader(new FileReader(file));
			while ((paraIdString = bReader.readLine()) != null)
			{
				paraIdString = paraIdString.trim();
				if (paraIdString.equals(""))
				{
					continue;
				}
				int paraId = Integer.valueOf(paraIdString);
				cfList.add(paraId);
			}
			
		}
		catch (IOException e1)
		{
			flag = false;
			logger.error("can not open cf file");
		}
		catch (NumberFormatException e2)
		{
			flag = false;
			logger.error("cf file contains invalid integer");
		}
		finally
		{
			if (flag == false)
			{
				logger.info("get cfs failed!");
			}
			try
			{
				if (bReader != null)
				{
					bReader.close();
				}
			}
			catch (IOException e)
			{
				logger.info(e.getMessage());
			}
		}
		if (flag == true)
		{
			cFS = cfList.toArray(new Integer[0]);
		}
		return flag;
	}
	
	protected static boolean getEquipIDs()
	{
		boolean flag = true;
		File file = new File(equipFile);
		BufferedReader bReader = null;
		String equipIdString = null;
		ArrayList<String> equipList = new ArrayList<String>();
		try
		{
			bReader = new BufferedReader(new FileReader(file));
			while ((equipIdString = bReader.readLine()) != null)
			{
				equipIdString = equipIdString.trim();
				if (equipIdString.equals(""))
				{
					continue;
				}
				equipList.add(equipIdString);
			}
			
		}
		catch (IOException e1)
		{
			flag = false;
			logger.error("can not open equip file");
		}
		finally
		{
			if (flag == false)
			{
				logger.info("get equip ids failed!");
			}
			try
			{
				if (bReader != null)
				{
					bReader.close();
				}
			}
			catch (IOException e)
			{
				logger.info(e.getMessage());
			}
		}
		if (flag == true)
		{
			equips = equipList.toArray(new String[0]);
		}
		return flag;
	}
	
	protected static boolean getQueryDateTimes()
	{
		boolean flag = true;
		boolean firstLine = true;
		File file = new File(querydateFile);
		BufferedReader bReader = null;
		String lineString = null;
		ArrayList<Date> startList = new ArrayList<Date>();
		ArrayList<Date> endList = new ArrayList<Date>();
		try
		{
			bReader = new BufferedReader(new FileReader(file));
			while ((lineString = bReader.readLine()) != null)
			{
				lineString = lineString.trim();
				String[] dateStrings = lineString.split(",");
				if (dateStrings.length != 2)
				{
					continue;
				}
				Date start, end;
				try
				{
					start = sdf.parse(dateStrings[0]);
					end = sdf.parse(dateStrings[1]);
				}
				catch (ParseException e)
				{
					continue;
				}
				if (start != null && end != null)
				{
					if (firstLine)
					{
						startGetQueryDate = start;
						endGetQueryDate = end;
						firstLine = false;
					}
					else
					{
						startList.add(start);
						endList.add(end);
					}
				}
			}
			
		}
		catch (IOException e1)
		{
			flag = false;
			logger.error("can not open querydate file");
		}
		finally
		{
			if (flag == false)
			{
				logger.info("get query dates failed!");
			}
			try
			{
				if (bReader != null)
				{
					bReader.close();
				}
			}
			catch (IOException e)
			{
				logger.info(e.getMessage());
			}
		}
		if (flag == true)
		{
			startQueryTimes = startList.toArray(new Date[0]);
			endQueryTimes = endList.toArray(new Date[0]);
		}
		return flag;
	}
}
